package com.lhj.queue.bootstrap;

import java.util.Calendar;
import java.util.Iterator;
import java.util.Objects;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lhj.queue.AbstractTask;
import com.lhj.queue.Slot;
import com.lhj.queue.WheelQueue;

/**
 * @ClassName QueueScanTimer
 * @Description 类似钟表的秒针，队列是表盘，这里有个类似秒针的循环器，
 *              每秒循环一次；就类似秒针再走。
 * @date 2017-03-21
 * @author hongjian.liu
 * @version 1.0.0
 *
 */
public class QueueScanTimer extends TimerTask {

	private static final Logger LOG = LoggerFactory.getLogger(QueueScanTimer.class);

	private static ThreadFactory slotThreadFactory = new NameThreadFactory("slot");
	private static ThreadFactory taskThreadFactory = new NameThreadFactory("task");

	/**处理每个槽位的线程，循环到这个槽位，立即丢到一个线程去处理，然后继续循环队列*/
	// private final ThreadPoolExecutor slotPool = new ThreadPoolExecutor(60, 60,
    //         0L, TimeUnit.MILLISECONDS,
    //         new LinkedBlockingQueue<>(1000),
	// 		slotThreadFactory);
	private final ExecutorService slotPool = Executors.newSingleThreadExecutor(slotThreadFactory);

	/**处理每一个槽位中task集合的线程， 集合中的每个任务一个线程*/
	private final ThreadPoolExecutor taskPool = new ThreadPoolExecutor(
			Runtime.getRuntime().availableProcessors(), Runtime.getRuntime().availableProcessors()*2 + 2,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(1000),
			taskThreadFactory);

	/**环形队列*/
	private final WheelQueue queue;

	private final long startTime;
	private int lastSecond = -1;

	public QueueScanTimer(WheelQueue queue) {
		Objects.nonNull(queue);
		this.queue = queue;
		this.startTime = Calendar.getInstance().getTimeInMillis();
	}

	@Override
	public void run() {
		try {
			Calendar calendar = Calendar.getInstance();
			int currentSecond = calendar.get(Calendar.MINUTE) * 60 + calendar.get(Calendar.SECOND);

			// QueueScanTimer.mockLongTime();

			checkMissSlotTask(currentSecond);
			LOG.info("lastSecond:{}, currentSecond(slot):{}", lastSecond, currentSecond);
			executeSlotTask(currentSecond);

			lastSecond = currentSecond;
		} catch (Exception e) {
			//这里一个槽位的屏蔽异常，继续执行。
			LOG.error(e.getMessage(), e);
		}
	}

	/** 检测丢失未执行的任务，并进行补充执行 */
	private void checkMissSlotTask(int currentSecond){
		if(lastSecond < 0){
			return;
		}

		//考虑定时器运行消耗的时间导致的误差，例如10:00:00,999，之后进位变成10:00:02,001
		//（1）发现由于定时器执行耗时，耗时累加导致的跨1秒问题，所以先执行跨过的那1秒，
		//可能这2秒的任务几乎同时执行，注意如果需要按顺序1秒1秒执行的话就保证不了了，目前设置slotPool池子只有1个，就按顺序执行了
		//（2）还有一种方案就是定时任务不按照1秒一次定时执行，而是按照500毫秒即0.5秒一次定时执行，即频率变快了，每1秒会执行2次
		int crossSecond = currentSecond - lastSecond;
		if(crossSecond > 1){ //距离上次运行时间（秒）超过1秒，就补充执行丢失的时间（秒）
			for(int missSecond=1;missSecond<crossSecond;missSecond++){
				int expectSecond = lastSecond + missSecond;
				LOG.info("lastSecond:{}, expectSecond(slot):{}, currentSecond(slot):{}", lastSecond, expectSecond, currentSecond);
				executeSlotTask(expectSecond);
			}
		}
	}

	private void executeSlotTask(int currentSecond){
		Slot slot = queue.peek(currentSecond);
		if(slot.hasTask()) {
			slotPool.execute(new SlotTask(slot.getTasks(), currentSecond));
		}
	}

	/**
	 * 槽位所有的任务
	 * @author hongjian.liu
	 *
	 */
    final class SlotTask implements Runnable {
    	final ConcurrentLinkedQueue<AbstractTask> tasks;
    	final int currentSecond;

		SlotTask(ConcurrentLinkedQueue<AbstractTask> tasks, int currentSecond) {
			this.tasks = tasks;
			this.currentSecond = currentSecond;
		}

		@Override
		public void run() {
			LOG.info("-----------------> currentSecond(slot):{}", currentSecond);

			// QueueScanTimer.mockLongTime();

			Iterator<AbstractTask> it = tasks.iterator();
            while (it.hasNext()) {
            	AbstractTask task = it.next();
            	if (LOG.isDebugEnabled()) {
            		LOG.debug("running_current_slot:currentSecond={}, task={}, taskQueueSize={}", currentSecond, task.toString(), tasks.size());
            	}
                if (task.canExecute()) {
                    taskPool.execute(task);
                    it.remove();
                    queue.removeAttr(task.getId());
                } else {
                	if (LOG.isDebugEnabled()) {
                		LOG.debug("countDown#running_current_slot:currentSecond={}, task={}", currentSecond, task.toString());
                	}
                    task.countDown();
                }
            }
		}
	}

	public static void mockLongTime(){
		try {
			Thread.sleep(1000 * 10);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
    
    /**
     * The default thread factory
     */
	static class NameThreadFactory implements ThreadFactory {
		// private static final AtomicInteger POOL_NUMBER = new AtomicInteger(1);
		private final ThreadGroup group;
		private final AtomicInteger threadNumber = new AtomicInteger(1);
		private final String namePrefix;

		NameThreadFactory(String groupName) {
			SecurityManager s = System.getSecurityManager();
			group = (s != null) ? s.getThreadGroup() :
					Thread.currentThread().getThreadGroup();
			namePrefix = groupName + /*"-pool-" + POOL_NUMBER.getAndIncrement() +*/ "-thread-";
		}

		@Override
		public Thread newThread(Runnable r) {
			Thread t = new Thread(group, r,
					namePrefix + threadNumber.getAndIncrement(),
					0);
			if (t.isDaemon()) {
				t.setDaemon(false);
			}
			if (t.getPriority() != Thread.NORM_PRIORITY) {
				t.setPriority(Thread.NORM_PRIORITY);
			}
			return t;
		}
	}

}
